package com.fulihui.wgj.consumer.service.impl;

import com.fulihui.common.util.PlatformUtil;

import com.fulihui.stater.kudu.client.KuduImpalaTemplate;
import com.fulihui.weixinsdk.WeixinSDK;
import com.fulihui.weixinsdk.model.UserInfo;
import com.fulihui.weixinsdk.model.UserOpenId;
import com.fulihui.weixinsdk.request.user.UserBatchGetRequest;
import com.fulihui.weixinsdk.request.user.UserInfoGetRequest;
import com.fulihui.weixinsdk.result.user.UserBatchGetResult;
import com.fulihui.weixinsdk.result.user.UserInfoGetResult;
import com.fulihui.wgj.consumer.entity.WxUserInfoExp;
import com.fulihui.wgj.consumer.service.RegionService;
import com.fulihui.wgj.consumer.service.UserService;
import com.fulihui.wgj.facade.dto.WxUserInfoDTO;
import com.fulihui.wgj.facade.msg.UserMsg;
import com.google.common.base.Joiner;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;

import org.apache.commons.lang3.StringUtils;
import org.apache.kudu.client.KuduException;
import org.apache.kudu.client.Operation;
import org.apache.kudu.client.Upsert;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.stream.Collectors;

/**
 * @author: LiuJie
 * @version: 2018/6/28 14:05
 * @description:
 */
@Service
@Slf4j
public class UserServiceImpl implements UserService{
    @Autowired
    private PlatformUtil platformUtil;
    @Autowired
    private WeixinSDK weixinSDK;
    @Autowired
    RegionService regionService;
    @Autowired
    private KuduImpalaTemplate kuduImpalaTemplate;


    @Override
    public List<WxUserInfoDTO> getUserInfo(UserMsg userMsg) {
        List<WxUserInfoDTO> list = new LinkedList<>();

        try {
            String accessToken = platformUtil.get(userMsg.getAppId());
            if (StringUtils.isNotBlank(accessToken)) {
                List<UserOpenId> userOpenIdList = userMsg.getOpenIdList().parallelStream()
                        .map(s -> {
                            UserOpenId openId = new UserOpenId();
                            openId.setOpenid(s);
                            return openId;
                        }).collect(Collectors.toList());
                // 微信接口一次最多拉取100条
                for (List<UserOpenId> openids : Lists.partition(userOpenIdList, 100)) {
                    list.addAll(requestUserInfo(userMsg.getAppId(), accessToken, openids));
                }
            }else {
                throw new RuntimeException("从 redis 中获取 access_token 失败");
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
        return list;
    }

    private  List<WxUserInfoDTO> requestUserInfo(String appid ,String accessToken, List<UserOpenId> openids) {
         List<WxUserInfoDTO> list = new LinkedList<>();
        UserBatchGetRequest userBatchGetRequest = new UserBatchGetRequest();
        userBatchGetRequest.setAccess_token(accessToken);
        userBatchGetRequest.setUser_list(openids);
        UserBatchGetResult userBatchGetResult = weixinSDK.invokeService(userBatchGetRequest);

        if (null != userBatchGetResult
                && !CollectionUtils.isEmpty(userBatchGetResult.getUser_info_list())) {
            userBatchGetResult.getUser_info_list().parallelStream()
                    .filter(userOpenId -> userOpenId.getSubscribe().equals(1L))
                    .forEach(userInfo -> {
                        WxUserInfoDTO userInfoDTO = new WxUserInfoDTO();
                        // 字段赋值
                        wxUserInfo2DbUserInfo(userInfo, userInfoDTO, appid);
                        list.add(userInfoDTO);
                    });
        }
        return list;
    }

    @Override
    public WxUserInfoExp getUserInfo(String appid, String openid) {
        String accessToken = platformUtil.get(appid);
        if (StringUtils.isBlank(accessToken)) {
            log.info("GetUserDetail!!get token is null");
            return null;
        }
        UserInfoGetRequest userGetRequest = new UserInfoGetRequest();
        userGetRequest.setAccess_token(accessToken);
        userGetRequest.setOpenid(openid);

        UserInfoGetResult userInfoGetResult = weixinSDK.invokeService(userGetRequest);
        UserInfo userInfo = userInfoGetResult.getUserInfo();
        WxUserInfoExp wxUserInfoExp = new WxUserInfoExp();
        wxUserInfo2DbUserInfo(userInfo, wxUserInfoExp, appid);
        // ---- 拓展字段
        wxUserInfoExp.setQrScene(userInfo.getQr_scene());
        wxUserInfoExp.setSubscribeScene(userInfo.getSubscribe_scene());
        return wxUserInfoExp;
    }

    @Override
    public void upsertUsers(List<WxUserInfoDTO> users) {
        if(CollectionUtils.isEmpty(users)){
            log.info("插入用户 list 为空");
            return;
        }
        try {
            doUpsert(users);
        }catch (KuduException e) {
            handleException(users, e);
        }
    }


    @Override
    public void delUser(String appid, String openId) {
        WxUserInfoDTO userInfoDTO = new WxUserInfoDTO();
        userInfoDTO.setOpenid(openId);
        try {
            kuduImpalaTemplate.delete(appid,userInfoDTO);
        } catch (KuduException e) {
            log.error("删除失败,appid="+appid+",openId="+openId,e);
        }
    }

    private void handleException(List<WxUserInfoDTO> users, KuduException e) {
        log.info("error msg={},e={}",e.getMessage(),e);
        //如果是表被删除则刷新表重试一次
        if(e.getMessage().contains("table was deleted")){
            Set<String> tables = new HashSet<>();
            for (WxUserInfoDTO user : users) {
                if(null != user){
                    tables.add(user.getAppId());
                }
            }
            reloadTable(tables);
            log.info("刷新表数据,tables={}",tables);
            try {
                // 重试
                doUpsert(users);
            } catch (KuduException e1) {
                log.error("刷新表后重新尝试，依旧抛异常。",e1);
            }
        }else {
            // 否则直接打印出来
            log.error("kudu插入用户异常,msg={}",e.getMessage());
        }
    }

    private void doUpsert(List<WxUserInfoDTO> users) throws KuduException {
        List<Operation> operations = new LinkedList<>();
        for (WxUserInfoDTO userInfoDTO : users) {

            if(null == userInfoDTO){
                continue;
            }
            Upsert upsert = kuduImpalaTemplate.createUpsert(userInfoDTO.getAppId(), userInfoDTO);
            operations.add(upsert);
        }
        kuduImpalaTemplate.apply(operations);
    }



    private void reloadTable(Set<String> tableNames){
        if(CollectionUtils.isEmpty(tableNames)){
            return;
        }
        try {
            for (String tableName : tableNames) {
                kuduImpalaTemplate.reloadTable(tableName);
            }
        } catch (KuduException e1) {
            e1.printStackTrace();
        }
    }

    /**
     * 将 微信SDK 类型的User 对应为 数据库类型
     * @param wxUserInfo 微信返回类型
     * @param dbUserInfo 数据库对象
     * @param appId
     */
    private void wxUserInfo2DbUserInfo(UserInfo wxUserInfo, WxUserInfoDTO dbUserInfo, String appId) {
        dbUserInfo.setAppId(appId);
        dbUserInfo.setOpenid(wxUserInfo.getOpenid());
        dbUserInfo.setUnionid(wxUserInfo.getUnionid());
        dbUserInfo.setSubscribe(wxUserInfo.getSubscribe());
        dbUserInfo.setNickname(wxUserInfo.getNickname());
        dbUserInfo.setSex(wxUserInfo.getSex());

        dbUserInfo.setCountry(regin2Rid(wxUserInfo.getCountry(), RegionService.DEFAULT_COUNTRY));
        dbUserInfo.setProvince(regin2Rid(wxUserInfo.getCountry()
                + wxUserInfo.getProvince(), RegionService.DEFAULT_PROVINCE));
        dbUserInfo.setCity(regin2Rid(wxUserInfo.getCountry()
                + wxUserInfo.getProvince()
                + wxUserInfo.getCity(), RegionService.DEFAULT_CITY));

        dbUserInfo.setLanguage(wxUserInfo.getLanguage());
        dbUserInfo.setHeadimgurl(wxUserInfo.getHeadimgurl());
        dbUserInfo.setSubscribeTime(wxUserInfo.getSubscribe_time() * 1000L);
        dbUserInfo.setRemark(wxUserInfo.getRemark());
        dbUserInfo.setGroupid(wxUserInfo.getGroupid());
        String tags = convertTags(wxUserInfo.getTagid_list());
        dbUserInfo.setTag(tags);
        dbUserInfo.setQrSceneStr(wxUserInfo.getQr_scene_str());
    }

    /**
     * 文字类型的国家省份信息 —》 rid
     * @param name 国家省份信息
     * @param defVal 当找不到该信息时的默认值
     * @return rid
     */
    private String regin2Rid(String name,String defVal){
        Map<String, String> regions = regionService.getRegion();
        return Optional.ofNullable(regions.get(name)).orElseGet(() -> regions.get(defVal));
    }

    private String convertTags(List<Integer> tagid_list) {
        String tags = Joiner.on("#").join(tagid_list);
        if (StringUtils.isNotEmpty(tags)) {
            tags = "#" + tags + "#";
        }
        return tags;
    }
}
